package io.debezium.connector.postgresql;

import io.debezium.pipeline.source.spi.EventMetadataProvider;
import io.debezium.pipeline.spi.OffsetContext;
import io.debezium.schema.DataCollectionId;
import io.debezium.time.Conversions;
import io.debezium.util.Collect;
import org.apache.kafka.connect.data.Struct;

import java.time.Instant;
import java.util.Map;

public class PostgresEventMetadataProvider implements EventMetadataProvider {
    PostgresEventMetadataProvider() {}

    public Instant getEventTimestamp(
            DataCollectionId source, OffsetContext offset, Object key, Struct value) {
        if (value == null) {
            return null;
        } else {
            Struct sourceInfo = value.getStruct("source");
            if (source == null) {
                return null;
            } else {
                Long timestamp;
                if (sourceInfo.schema().field("ts_usec") != null) {
                    timestamp = sourceInfo.getInt64("ts_usec");
                    return timestamp == null ? null : Conversions.toInstantFromMicros(timestamp);
                } else {
                    timestamp = sourceInfo.getInt64("ts_ms");
                    return timestamp == null ? null : Instant.ofEpochMilli(timestamp);
                }
            }
        }
    }

    public Map<String, String> getEventSourcePosition(
            DataCollectionId source, OffsetContext offset, Object key, Struct value) {
        if (value == null) {
            return null;
        } else {
            Struct sourceInfo = value.getStruct("source");
            if (source == null) {
                return null;
            } else {
                Long xmin = sourceInfo.getInt64("xmin");
                Long lsn = sourceInfo.getInt64("lsn");
                if (lsn == null) {
                    return null;
                } else {
                    Map<String, String> r = Collect.hashMapOf("lsn", Long.toString(lsn));
                    if (xmin != null) {
                        r.put("xmin", Long.toString(xmin));
                    }

                    return r;
                }
            }
        }
    }

    public String getTransactionId(
            DataCollectionId source, OffsetContext offset, Object key, Struct value) {
        if (value == null) {
            return null;
        } else {
            Struct sourceInfo = value.getStruct("source");
            if (source == null) {
                return null;
            } else {
                Long txId = sourceInfo.getInt64("txId");
                return txId == null ? null : Long.toString(txId);
            }
        }
    }
}
